package top.doe.spark.netty

import top.doe.spark.rpc.{AbortableRpcFuture, RpcEndpointAddress, RpcEndpointRef, RpcTimeout}
import top.doe.spark.util.RpcAddress

import scala.concurrent.Future
import scala.reflect.ClassTag

class NettyRpcEndpointRef(
  endpointAddress: RpcEndpointAddress,
  var nettyEnv: NettyRpcEnv
) extends RpcEndpointRef{

  override def name: String = endpointAddress.name

  override def address: RpcAddress = endpointAddress.rpcAddress

  //发送异步消息
  override def send(message: Any): Unit = {
    nettyEnv.send(new RequestMessage(nettyEnv.address, this, message))
  }

  //ask 发送同步消息
  override def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T] = {
    askAbortable(message, timeout).future
  }

  //发送同步请求，如果出现异常，对异常进行处理
  override def askAbortable[T: ClassTag](message: Any, timeout: RpcTimeout): AbortableRpcFuture[T] = {
    //使用nettyEvn发消息
    nettyEnv.askAbortable(new RequestMessage(nettyEnv.address, this, message), timeout)
  }
}

class RequestMessage(
  val senderAddress: RpcAddress,
  val receiver: NettyRpcEndpointRef,
  val content: Any
) {

  //发给远端的消息，要先序列化
  //TODO 远端的消息，发送前要先序列化
}
